# import ipaddress
# import threading
# import warnings
# import paramiko
# from ccolor import Colors
# # import logging
# # logging.getLogger("paramiko").setLevel(logging.CRITICAL)
# warnings.filterwarnings("ignore")
# warnings.simplefilter("ignore")
#
#
# def ssh_run(host, port, users, passwds):
#     """
#     SSH弱口令扫描主函数
#     :param host: 目标主机
#     :param port: SSH端口
#     :param users: 用户名列表
#     :param passwds: 密码列表
#     """
#     print(f"{Colors.YELLOW}[*] 开始扫描SSH弱口令: {host}:{port}{Colors.RESET}")
#
#     # 预处理用户名和密码，去除空白和空行
#     user_list = [user.strip() for user in users if user.strip()]
#     passwd_list = [passwd.strip() for passwd in passwds if passwd.strip()]
#
#     if not user_list or not passwd_list:
#         print(f"{Colors.RED}[-] 用户名或密码列表为空{Colors.RESET}")
#         return
#
#     print(f"{Colors.CYAN}[*] 加载用户名: {len(user_list)} 个, 密码: {len(passwd_list)} 个{Colors.RESET}")
#
#     # 控制并发线程数量
#     max_threads = 300
#     semaphore = threading.Semaphore(max_threads)
#
#
#     if "/" in host:
#         threads = []
#         for ip in ipaddress.IPv4Network(host):
#             ip = str(ip)
#             for user in user_list:
#                 for passwd in passwd_list:
#                     t = threading.Thread(target=ssh_connect,args=(ip, port, user, passwd, semaphore))
#                     t.start()
#                     threads.append(t)
#         for t in threads:
#             t.join()
#     else:
#         threads = []
#         for user in user_list:
#             for passwd in passwd_list:
#                 t = threading.Thread(
#                     target=ssh_connect,
#                     args=(host, port, user, passwd, semaphore)
#                 )
#                 t.start()
#                 threads.append(t)
#         for t in threads:
#                 t.join()
#     # 等待所有线程完成
#
#
#     print(f"{Colors.YELLOW}[*] SSH弱口令扫描完成{Colors.RESET}")
#
#
# def ssh_connect(host, port, username, password, semaphore):
#     """
#     SSH连接测试函数
#     """
#     semaphore.acquire()
#     try:
#         ssh = paramiko.SSHClient()
#         # 自动添加主机密钥
#         ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
#
#         # 设置连接超时和认证超时
#         ssh.connect(
#             host,
#             port=port,
#             username=username,
#             password=password,
#             timeout=1,
#             banner_timeout=1,
#             auth_timeout=1
#         )
#
#         # 验证连接是否成功
#         if ssh.get_transport().is_active():
#             # 执行简单命令验证
#             stdin, stdout, stderr = ssh.exec_command("echo lyf", timeout=5)
#             output = stdout.read().decode('utf-8').strip()
#
#             if "lyf" in output:
#                 print(f"{Colors.GREEN}[+] {host}:{port} SSH存在弱口令 [{username} / {password}]{Colors.RESET}")
#
#     except paramiko.AuthenticationException:
#         # 认证失败是正常情况，无需处理
#         pass
#     except paramiko.SSHException:
#         pass
#     except Exception:
#         pass
#     finally:
#         try:
#             ssh.close()
#         except:
#             pass
#         semaphore.release()


# import ipaddress
# import threading
# import paramiko
# from ccolor import Colors
# import warnings
#
# warnings.filterwarnings("ignore")
# warnings.simplefilter("ignore")
#
#
# def ssh_run(host, port, users, passwds):
#     """
#     SSH弱口令扫描主函数
#     :param host: 目标主机/IP段
#     :param port: SSH端口
#     :param users: 用户名列表
#     :param passwds: 密码列表
#     """
#     print(f"{Colors.YELLOW}[*] 开始扫描SSH弱口令: {host}:{port}{Colors.RESET}")
#
#     # 预处理用户名和密码
#     users = [u.strip() for u in users if u.strip()]
#     passwds = [p.strip() for p in passwds if p.strip()]
#
#     if not users or not passwds:
#         print(f"{Colors.RED}[-] 用户名或密码列表为空{Colors.RESET}")
#         return
#
#     print(f"{Colors.CYAN}[*] 用户名: {len(users)} 个, 密码: {len(passwds)} 个{Colors.RESET}")
#
#     # 控制并发线程数量
#     max_threads = 100
#     semaphore = threading.Semaphore(max_threads)
#
#     threads = []
#     if '/' in host:
#         # 网段扫描模式
#         try:
#             network = ipaddress.IPv4Network(host, strict=False)
#             print(f"{Colors.CYAN}[*] 扫描网段: {network}{Colors.RESET}")
#
#             for ip in network.hosts():
#                 ip_str = str(ip)
#                 for user in users:
#                     for passwd in passwds:
#                         t = threading.Thread(
#                             target=ssh_connect,
#                             args=(ip_str, port, user, passwd, semaphore)
#                         )
#                         t.start()
#                         threads.append(t)
#         except ValueError:
#             print(f"{Colors.RED}[-] 无效的网段格式: {host}{Colors.RESET}")
#             return
#     else:
#         # 单IP扫描模式
#         for user in users:
#             for passwd in passwds:
#                 t = threading.Thread(
#                     target=ssh_connect,
#                     args=(host, port, user, passwd, semaphore)
#                 )
#                 t.start()
#                 threads.append(t)
#
#     # 等待所有线程完成
#     for t in threads:
#         t.join()
#
#     print(f"{Colors.YELLOW}[*] SSH弱口令扫描完成{Colors.RESET}")
#
#
# def ssh_connect(host, port, username, password, semaphore):
#     """
#     SSH连接测试函数
#     """
#     semaphore.acquire()
#     try:
#         ssh = paramiko.SSHClient()
#         ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
#
#         # 优化超时设置
#         ssh.connect(
#             host,
#             port=port,
#             username=username,
#             password=password,
#             timeout=5,
#             banner_timeout=10,
#             auth_timeout=5
#         )
#
#         if ssh.get_transport().is_active():
#             stdin, stdout, stderr = ssh.exec_command("echo lyf", timeout=3)
#             output = stdout.read().decode('utf-8').strip()
#
#             if "lyf" in output:
#                 print(f"{Colors.GREEN}[+] {host}:{port} SSH存在弱口令 [{username}/{password}]{Colors.RESET}")
#
#     except paramiko.AuthenticationException:
#         pass  # 认证失败是正常情况
#     except Exception:
#         pass  # 静默处理所有其他异常
#     finally:
#         try:
#             ssh.close()
#         except:
#             pass
#         semaphore.release()
#
#
# if __name__ == "__main__":
#     # 测试代码
#     test_users = ["root", "admin"]
#     test_passwds = ["123456", "password"]
#
#     # 测试单IP扫描
#     ssh_run("127.0.0.1", 22, test_users, test_passwds)
#
#     # 测试网段扫描（取消注释测试）
#     # ssh_run("192.168.1.0/24", 22, test_users, test_passwds)

import ipaddress
import threading
import paramiko
from ccolor import Colors
import warnings

warnings.filterwarnings("ignore")
warnings.simplefilter("ignore")


def ssh_run(host, port, users, passwds):
    """
    SSH弱口令扫描主函数
    :param host: 目标主机/IP段
    :param port: SSH端口
    :param users: 用户名列表
    :param passwds: 密码列表
    """
    print(f"{Colors.YELLOW}[*] 开始扫描SSH弱口令: {host}:{port}{Colors.RESET}")

    # 预处理用户名和密码
    users = [u.strip() for u in users if u.strip()]
    passwds = [p.strip() for p in passwds if p.strip()]

    if not users or not passwds:
        print(f"{Colors.RED}[-] 用户名或密码列表为空{Colors.RESET}")
        return

    print(f"{Colors.CYAN}[*] 用户名: {len(users)} 个, 密码: {len(passwds)} 个{Colors.RESET}")

    # 控制并发线程数量
    max_threads = 100
    semaphore = threading.Semaphore(max_threads)

    threads = []
    if '/' in host:
        # 网段扫描模式
        try:
            network = ipaddress.IPv4Network(host, strict=False)
            print(f"{Colors.CYAN}[*] 扫描网段: {network}{Colors.RESET}")

            for ip in network.hosts():
                ip_str = str(ip)
                for user in users:
                    for passwd in passwds:
                        t = threading.Thread(
                            target=ssh_connect,
                            args=(ip_str, port, user, passwd, semaphore)
                        )
                        t.start()
                        threads.append(t)
        except ValueError:
            print(f"{Colors.RED}[-] 无效的网段格式: {host}{Colors.RESET}")
            return
    else:
        # 单IP扫描模式
        for user in users:
            for passwd in passwds:
                t = threading.Thread(
                    target=ssh_connect,
                    args=(host, port, user, passwd, semaphore)
                )
                t.start()
                threads.append(t)

    # 等待所有线程完成
    for t in threads:
        t.join()

    print(f"{Colors.YELLOW}[*] SSH弱口令扫描完成{Colors.RESET}")


def ssh_connect(host, port, username, password, semaphore):
    """
    SSH连接测试函数
    """
    semaphore.acquire()
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        # 优化超时设置
        ssh.connect(
            host,
            port=port,
            username=username,
            password=password,
            timeout=5,
            banner_timeout=10,
            auth_timeout=5
        )

        if ssh.get_transport().is_active():
            stdin, stdout, stderr = ssh.exec_command("echo lyf", timeout=3)
            output = stdout.read().decode('utf-8').strip()

            if "lyf" in output:
                print(f"{Colors.GREEN}[+] {host}:{port} SSH存在弱口令 [{username}/{password}]{Colors.RESET}")

    except paramiko.AuthenticationException:
        pass  # 认证失败是正常情况
    except Exception:
        pass  # 静默处理所有其他异常
    finally:
        try:
            ssh.close()
        except:
            pass
        semaphore.release()


if __name__ == "__main__":
    # 测试代码
    test_users = ["root", "admin"]
    test_passwds = ["123456", "password"]

    # 测试单IP扫描
    ssh_run("127.0.0.1", 22, test_users, test_passwds)

    # 测试网段扫描（取消注释测试）
    # ssh_run("192.168.1.0/24", 22, test_users, test_passwds)


